package com.hs.kfk.intercepter;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @Description: 自定义发送消息拦截器
 * @Author 胡尚
 * @Date: 2024/8/8 12:12
 */
public class MyInterceptor implements ProducerInterceptor {

    @Override
    public ProducerRecord onSend(ProducerRecord record) {
        System.out.println("send(ProducerRecord, Callback)方法，在key和value被序列化和分配分区partition之前调用");
        // 我们可以对record做相应的处理
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        System.out.println("当发送到服务器的记录已被确认/发送记录在发送到服务器之前失败/KafkaProducer.send()抛出异常时也会调用。");
    }

    @Override
    public void close() {
        System.out.println("连接关闭时会被调用");
    }

    @Override
    public void configure(Map<String, ?> configs) {
        System.out.println("整理配置项");
        System.out.println("=====config start======");
        for (Map.Entry<String, ?> entry : configs.entrySet()) {
            System.out.println("entry.key:"+entry.getKey()+"\t === entry.value: "+entry.getValue());
        }
        System.out.println("=====config end======");

    }
}
